package com.atguigu.tingshu.album.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.vo.album.CanalListenAlbumStatMsgVo;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.List;

/**
 * Author  王东剑
 * Created on 19/6/2024 PM8:58
 * Package_name com.atguigu.tingshu.album.canal
 * Description
 */
@CanalEventListener
@Slf4j
public class CustomCanalClientListener {
	@Autowired
	private StringRedisTemplate redisTemplate;
	@Autowired
	private KafkaService kafkaService;

	// 监听专辑的变化
	@ListenPoint(destination = "example", schema = "tingshu_album", table = {"album_info"})
	public void listenerAlbumInfo(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
		log.info("CanalClientListener.listenerAlbumInfo() 方法执行中........");
		List<CanalEntry.Column> columnsList = rowData.getAfterColumnsList();
		columnsList.forEach(column -> {
			String columnName = column.getName();
			String columnValue = column.getValue();
			if ("id".equals(columnName)) {
				log.info("检测到 album_info 更改,删除Redis中的数据: {}", RedisConstant.ALBUM_INFO_PREFIX + columnValue);
				redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX + columnValue);
			}
		});
	}

	// 	监听专辑统计信息的变化
	@SneakyThrows
	@ListenPoint(destination = "example", schema = "tingshu_album", table = {"album_stat"})
	public void listenerAlbumStat(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
		CanalListenAlbumStatMsgVo.CanalListenAlbumStatMsgVoBuilder builder = CanalListenAlbumStatMsgVo.builder();
		log.info("CanalClientListener.listenerAlbumStat() 方法执行中........");
		List<CanalEntry.Column> columnsList = rowData.getAfterColumnsList();
		columnsList.forEach(column -> {
			String columnName = column.getName();
			String columnValue = column.getValue();
			if ("album_id".equals(columnName)) {
				builder.albumId(Long.parseLong(columnValue));
				log.info("检测到 album_stat 更改,删除 Redis 中的数据: {}", RedisConstant.ALBUM_STAT_PREFIX + columnValue);
				redisTemplate.delete(RedisConstant.ALBUM_STAT_PREFIX + columnValue);
			}
			if ("stat_type".equals(columnName)) {
				builder.statType(columnValue);
			}
			if ("stat_num".equals(columnName)) {
				builder.statNum(Integer.parseInt(columnValue));

			}
		});
		CanalListenAlbumStatMsgVo canalListenAlbumStatMsgVo = builder.build();
		log.info("将 album_stat 的更改数据添加到 Kafka队列: {}", canalListenAlbumStatMsgVo);
		kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_STAT_CANAL_TO_SEARCH, JSON.toJSONString(canalListenAlbumStatMsgVo));
	}

	// 	监听专辑统计信息的变化
	@SneakyThrows
	@ListenPoint(destination = "example", schema = "tingshu_user", table = {"user_info"})
	public void listenerUserInfo(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
		log.info("CanalClientListener.listenerUserInfo() 方法执行中........");
		List<CanalEntry.Column> columnsList = rowData.getAfterColumnsList();
		columnsList.forEach(column -> {
			String columnName = column.getName();
			String columnValue = column.getValue();
			if ("id".equals(columnName)) {
				log.info("检测到 user_info 更改,删除 Redis 中的数据: {}", RedisConstant.CACHE_USER_PREFIX + columnValue);
				redisTemplate.delete(RedisConstant.CACHE_USER_PREFIX + columnValue);
			}
		});
	}
}
